package cn.bytengine.stem.cluster.udp;

import cn.bytengine.stem.module.AbstractModule;
import cn.hutool.core.lang.Tuple;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.DatagramPacket;
import io.netty.channel.socket.InternetProtocolFamily;
import io.netty.channel.socket.nio.NioDatagramChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.NetworkInterface;
import java.nio.charset.StandardCharsets;

import static cn.bytengine.stem.utils.Address.localAddress;


public class NettyUdpModule extends AbstractModule {
    private static final Logger log = LoggerFactory.getLogger(NettyUdpModule.class);
    public static final String MODULE_NAME = "stem-cluster-netty-udp-socket";
    private final ClusterUdpProps props;

    public NettyUdpModule(ClusterUdpProps props) {
        super(MODULE_NAME);
        this.props = props;
    }

    @Override
    protected void init() {
        Tuple result = localAddress();
        if (result == null) {
            throw new IllegalStateException("Not found allow IPv4 address for host.");
        }
        NetworkInterface ni = result.get(0);
        InetAddress localAddress = result.get(1);
        // launch udp server
        boolean launchResult = startUdpServe(ni, localAddress, props);
        // launch client to send group packet
        if (launchResult) {
            sendJoinMessage(ni, localAddress, props);
        }
    }

    protected boolean startUdpServe(NetworkInterface ni,
                                    InetAddress localAddress,
                                    ClusterUdpProps props) {
        final int port = props.getListenPort();
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        b.group(group)
                .channelFactory((ChannelFactory<NioDatagramChannel>) () -> new NioDatagramChannel(InternetProtocolFamily.IPv4))
                .localAddress(localAddress, port)
                .option(ChannelOption.IP_MULTICAST_IF, ni)
                .option(ChannelOption.SO_REUSEADDR, true)
                .handler(new ChannelInitializer<NioDatagramChannel>() {
                    @Override
                    public void initChannel(NioDatagramChannel ch) throws Exception {
                        ch.pipeline().addLast(new ServerMulticastHandler());
                    }
                });
        try {
            NioDatagramChannel ch = (NioDatagramChannel) b.bind(port).sync().channel();
            ch.joinGroup(new InetSocketAddress(props.getMulticastGroup(), port), ni).sync();
            log.info("Listen succeeded");
            return true;
        } catch (Exception ex) {
            log.error("launching udp server occurred error.", ex);
            return false;
        }
    }

    protected boolean sendJoinMessage(NetworkInterface ni,
                                      InetAddress localAddress,
                                      ClusterUdpProps props) {
        final int port = props.getListenPort();
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        b.group(group)
                .channelFactory((ChannelFactory<NioDatagramChannel>) () -> new NioDatagramChannel(InternetProtocolFamily.IPv4))
                .localAddress(localAddress, port)
                .option(ChannelOption.IP_MULTICAST_IF, ni)
                .option(ChannelOption.SO_REUSEADDR, true)
                .handler(new ChannelInitializer<NioDatagramChannel>() {
                    @Override
                    public void initChannel(NioDatagramChannel ch) throws Exception {
                        ch.pipeline().addLast(new ClientMulticastHandler());
                    }
                });
        try {
            Channel ch = b.bind().sync().channel();
            ch.writeAndFlush(new DatagramPacket(
                    Unpooled.copiedBuffer("1", StandardCharsets.UTF_8),
                    new InetSocketAddress(props.getMulticastGroup(), port))).sync();

            ch.close().awaitUninterruptibly();
            log.info("sent join message succeeded");
            return true;
        } catch (Exception ex) {
            log.error("sent join message occurred error.", ex);
            return false;
        } finally {
            group.shutdownGracefully();
        }
    }
}

class ClientMulticastHandler extends SimpleChannelInboundHandler<DatagramPacket> {
    private static final Logger log = LoggerFactory.getLogger(ClientMulticastHandler.class);

    @Override
    public void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) throws Exception {
        String response = msg.content().toString(StandardCharsets.UTF_8);
        log.info("client receive message from server: {}", response);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        String channelId = ctx.channel().id().asShortText();
        log.error("The channel '{}' uncaught exception: ", channelId, cause);
        ctx.close();
    }
}

class ServerMulticastHandler extends SimpleChannelInboundHandler<DatagramPacket> {
    private static final Logger log = LoggerFactory.getLogger(ServerMulticastHandler.class);

    @Override
    public void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) throws Exception {
        log.info("receive content: '{}', from {}", packet.content().toString(StandardCharsets.UTF_8), packet.sender().getHostString());
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
        ctx.close();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        String channelId = ctx.channel().id().asShortText();
        log.error("The channel '{}' uncaught exception: ", channelId, cause);
    }
}
